<?php
declare(strict_types=1);

namespace ServerApp\lib\process;

use App\Lib\SwoftUtil;
use Co;
use ServerApp\lib\memory\Chan;
use Swoole\Process;
use Throwable;
use txz\common\base\utils\LogHelper;
use txz\common\dao\base\mysql\CMysqlMgr;
use WebApp\lib\util\Util;

/**
 * Class SProcess
 * @package txz\common\third_lib\model\process
 */
class SProcess extends Process
{
    /**
     * @param mixed $task
     * @param bool $redirect_pip_output
     * @param bool|int $createPip
     * @param null $name
     * @internal param mixed $callback 子进程的回调函数
     * @internal param bool $redirect_stdin_stdout 是否重定向标准输入输出
     * @internal param bool $create_pipe 是否创建管道
     */
    public function __construct($task, $redirect_pip_output = false, $createPip = 0, $name = '')
    {
        $func = function (Process $process) use ($task, $name) {
            $GLOBALS['swoole_pid'] = $process->pid;
            SGlobal::getInstance()->process = $process;
            if ($name) {
                swoole_set_process_name($name);
            }
            if ($task instanceof SPRunnable) {
                $task->run();
            } else {
                $task($process);
            }
        };
        parent::__construct($func, $redirect_pip_output, $createPip);
    }

    private $using = 0;

    /**
     * @param $string
     */
    public function work($string = ''): void
    {
        $this->using = 1;
        $this->write($string);
    }

    public function idel()
    {
        $this->using = 0;
    }

    /**
     * @return bool
     */
    public function isUsing()
    {
        return !!$this->using;
    }

    /**
     * 启动子进程
     *
     * @return int
     */
    public function start()
    {
        $pid = parent::start();
        SGlobal::getInstance()->child_process_pids[$pid] = null;
        return $pid;
    }

    /**
     * @param $process_name
     * @param $worker_num
     * @param $maker
     * @param $handler
     * @param callable $start
     * @param callable $end
     * @param int $coroutine_num
     */
    public static function createCoroutinePool($process_name, $worker_num, $maker, $handler, $start = null, $end = null,
                                            $coroutine_num = 1)
    {
        $process_name = '__' . $process_name;
        CMysqlMgr::clear(true);
//        foreach ($GLOBALS as $key => $val) {
//            if (Util::startWitsh('mysqlPool', $key)) {
//                unset($GLOBALS[$key]);
//            }
//        }

        $c5 = chr(5);
        $c6 = chr(6);
        $c7 = chr(7);
        $c8 = chr(8);
        $c9 = $c7 . $c8;

        pcntl_signal(SIGCHLD, SIG_IGN);
        $cmd = "ps -ef | grep '$process_name' | grep -v grep | cut -c 9-15 | xargs kill -s 9";
        shell_exec($cmd);
        Chan::getInstance($process_name . '_chan', 100 * 100 * 100 * 100 * 4);

        $pid_arr = [];
        for ($i = 0; $i < $worker_num; $i++) {
            $the_process_name = $process_name . '_' . $i;
            $pid_arr[] = (new SProcess(function ($p) use (
                &$c5, &$c6, &$c9,
                &$handler, &$process_name, &$the_process_name, &$start, &$end, &$coroutine_num, $i
            ) {
//                ServerStation::$wId = 111;
                if ($start) {
                    $start($the_process_name, $i);
                }

                go(function() use(
                    &$c5, &$c6, &$c9, &$p,
                    &$handler, &$process_name, &$the_process_name, &$end, &$coroutine_num, $i
                ) {
                    Co::set([
                        'max_coroutine' => $coroutine_num + 2,
                    ]);

                    $max_coroutine_num = $coroutine_num;
                    $static_index = 0;
                    while (true) {
                        $line = Chan::getInstance($process_name . '_chan')->pop();
                        if ($line) {
                            if ($line === $c9) {
                                break;
                            } else {
                                if ($line[0] === $c5) {
                                    $params = explode($c6, substr($line, 1, -1));
                                } else {
                                    $params = [substr($line, 0, -1)];
                                }

                                $params[] = $static_index++;

                                while (true) {
                                    if ($max_coroutine_num <= 0) {
                                        usleep(10000);
                                        continue;
                                    }
                                    $max_coroutine_num--;

                                    $res = go(function() use(
                                        &$params,
                                        &$handler,
                                        $i,
                                        &$max_coroutine_num
                                    ) {
                                        //*****************************************************************************
                                        $params[] = $i;
                                        $params[] = $max_coroutine_num;
                                        try {
                                            $handler(...$params);
                                        } catch (Throwable $exception) {
                                            Util::logE(Util::err($exception));
                                        }
                                        //*****************************************************************************
                                        $max_coroutine_num++;
                                    });
                                    if (!$res) {
                                        Util::logE("UUUUUUUUUUUUUUUUUU");
                                    }
                                    break;
                                }
                            }
                        } else {
                            usleep(10000);
                        }
                    }

                    var_dump('wait');
                    $func = function() use(&$coroutine_num, &$the_process_name, &$end, &$p, &$max_coroutine_num) {
                        if ($max_coroutine_num === $coroutine_num) {
                            if ($end) {
                                $end($the_process_name);
                            }
                            SProcess::kill($p->pid, 9);
                        }
                    };
                    $func();
                    swoole_timer_tick(1000, $func);
                });
            }, false, 0, $the_process_name))->start();
        }

        foreach ($maker() as $data) {
            if (is_array($data)) {
                $data = $c5 . join($c6, $data);
            }
            Chan::getInstance($process_name . '_chan')->push($data . $c5);
        }

        for ($i = 0; $i < $worker_num; $i++) {
            Chan::getInstance($process_name . '_chan')->push($c9);
        }

        while (true) {
            $count = (int)shell_exec("ps -ef | grep $process_name | grep -v grep | wc -l");
            if ($count === 0) {
                break;
            }
            sleep(2);
        }
    }

    public static function getLogTag()
    {
        return Util::get_global("task_process_name");
    }

    public static function setLogTag($val)
    {
        Util::set_global("task_process_name", $val);
    }

    /**
     * @param $process_name
     * @param $worker_num
     * @param $maker
     * @param $handler
     * @param callable $start
     * @param callable $end
     */
    public static function createWorkerPool($process_name, $worker_num, $maker, $handler, $start = null, $end = null)
    {
        $process_name = '__' . $process_name;

        CMysqlMgr::clear(true);
        SwoftUtil::clean();

        foreach ($GLOBALS as $key => $val) {
            if (Util::startWitsh('mysqlPool', $key)) {
                unset($GLOBALS[$key]);
            }
        }

        $c5 = chr(5);
        $c6 = chr(6);
        $c7 = chr(7);
        $c8 = chr(8);
        $c9 = $c7 . $c8;

        pcntl_signal(SIGCHLD, SIG_IGN);
        $cmd = "ps -ef | grep '$process_name' | grep -v grep | cut -c 9-15 | xargs kill -s 9";
        shell_exec($cmd);
        Chan::getInstance($process_name . '_chan', 100 * 100 * 100 * 100 * 4);

        $process_arr = [];
        $pid_dict = [];
        for ($i = 0; $i < $worker_num; $i++) {
            $the_process_name = $process_name . '_' . $i;
            $process = (new SProcess(function (\swoole_process $p) use (
                $c5, $c6, $c9,
                $handler, $process_name, $the_process_name, $start, $end) {

                self::setLogTag($the_process_name);

                CMysqlMgr::clear(true);
                SwoftUtil::clean();

                if ($start) {
                    $start($the_process_name);
                }

                $static_index = 0;
                while (true) {
                    $line = Chan::getInstance($process_name . '_chan')->pop();
                    if ($line) {
                        if ($line === $c9) {
                            if ($end) {
                                $end($the_process_name);
                            }

                            Util::logD("$the_process_name end write end");
                            $p->write("end");
                            SProcess::kill($p->pid, 9);
                        } else {
                            if ($line[0] === $c5) {
                                $params = explode($c6, substr($line, 1, -1));
                            } else {
                                $params = [substr($line, 0, -1)];
                            }

                            $params[] = $static_index++;

//                            usleep(random_int(10000, 100000));
                            $handler(...$params);
                        }
                    } else {
                        usleep(10000);
                    }
                }
            }, false, 2, $the_process_name));
            $process_arr[] = $process;
            $pid = $process->start();
            Util::logD("new process($i) pid=$pid");
            $pid_dict[$pid] = 1;
        }

        foreach ($maker() as $data) {
            if (is_array($data)) {
                $data = $c5 . join($c6, $data);
            }
            try {
                Chan::getInstance($process_name . '_chan')->push($data . $c5);
            } catch (Throwable $e) {
                Util::logObj($data, "chan push err = ");
                if (is_array($data)) {
                    LogHelper::logD("chan push err = " . json_encode($data));
                } else {
                    LogHelper::logD("chan push err = " . strlen($data));
                }
            }
        }

        for ($i = 0; $i < $worker_num; $i++) {
            Chan::getInstance($process_name . '_chan')->push($c9);
        }

        while (true) {
            usleep(100000);

            foreach ($process_arr as $process) {
                $key = $process->read();
                if ($key === "end") {
                    $pid_dict[$process->pid] = 0;
                }
            }

            $all_end = true;
            foreach ($pid_dict as $pid => $running) {
                if ($running) {
                    $all_end = false;
                    break;
                }
            }

            if ($all_end) {
                break;
            }

//            $count = (int)shell_exec("ps -ef | grep $process_name | grep -v grep | wc -l");
////            Util::logD($count);
//            if ($count === 0) {
//                break;
//            }
        }
    }
}
